package com.world.rickcloudendpoint.demo.listener;

import com.world.rickcloudendpoint.demo.QueueConstantKey;
import com.world.rickcloudendpoint.demo.RedisService;
import com.world.rickcloudendpoint.demo.MsgConsumer;
import com.world.rickcloudendpoint.demo.model.DemoMessage;
import com.world.rickcloudendpoint.demo.redis.RedisMqConsumerContainer;
import lombok.extern.log4j.Log4j2;

import java.util.concurrent.TimeUnit;

/**
 * @description: 消息队列监听
 * @author: jlsong
 * @date: 2022/03/17 21:37:13
 * @version: 1.0
 */
@Log4j2
public class RedisQueueListener implements Runnable {

    private RedisService redisService;
    private String queueName;
    private MsgConsumer consumer;

    public RedisQueueListener(RedisService redisService, String queueName, MsgConsumer consumer) {
        this.redisService = redisService;
        this.queueName = queueName;
        this.consumer = consumer;
    }

    @Override
    public void run() {
        log.info("Redis queue listener 正在启动...queue: {} ", queueName);
        while (RedisMqConsumerContainer.run) {
            try {
                // brpop 阻塞到有消息来才拉取消息
//                Object msg = redisTemplate.boundListOps(queueName).rightPop();
                String msg = redisService.lBRightPop(queueName, 1, TimeUnit.SECONDS);
                if (msg == null || msg.trim().length() == 0) {
                    continue;
                }

                DemoMessage demoMessage = QueueConstantKey.om.readValue(msg, DemoMessage.class);
                // 处理消息
                try {
                    consumer.onMessage(demoMessage);
                } catch (Exception e) {
                    consumer.onError(demoMessage, e);
                }
            } catch (Exception e) {
                if (RedisMqConsumerContainer.run) {
                    log.error("异常队列：{},原因：{}", queueName, e.getMessage(), e);
                } else {
                    log.info("Redis queue listener 退出.....queue: {} ,原因：{}", queueName, e.getMessage(), e);
                }
            }
        }
    }
}
